package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo10ReduceByKey {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo10ReduceByKey")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)
    // 统计性别人数
    val stuLineRDD: RDD[String] = sc.textFile("Spark/data/students.txt")
    // 将数据变成KV格式 以性别作为Key 1作为Value
    val genderKVRDD: RDD[(String, Int)] = stuLineRDD.map(line => (line.split(",")(3), 1))

    val reduceRDD: RDD[(String, Int)] = genderKVRDD

      /**
       * reduceByKey：转换算子
       * 需要接收一个函数f：(Int,Int) => Int 相当于需要传入一个聚合函数
       * 然后会对RDD的Key进行分组，并将聚合函数作用在每个分组上，相当于分组聚合操作
       * 思考：
       * reduceByKey vs groupByKey 区别？
       * 1、groupByKey只会对Key进行分组，reduceByKey不仅会对Key进行分组，还会进行聚合操作
       * 2、reduceByKey会在Map端进行预聚合（相当于MR中的Combiner操作），可以减少shuffle过程中传输的数据量进而提高效率
       * 3、reduceByKey不能做avg的操作，只能做sum、max、min、count操作（幂等操作）
       *
       */
      //      .reduceByKey((i1, i2) => i1 + i2)
      .reduceByKey(_ + _) // 省略的写法

    reduceRDD
//      .foreach(println)


    // 统计班级的平均年龄
    // 以班级作为Key，年龄作为Value 构建KV格式的RDD
    val clazzAgeKVRDD: RDD[(String, Double)] = stuLineRDD
      .map(line => (line.split(",")(4), line.split(",")(2).toDouble))

    // groupByKey
    clazzAgeKVRDD
      .groupByKey()
      .map(kv => {
        s"${kv._1},${kv._2.sum / kv._2.size}"
      })
      .foreach(println)

    // reduceByKey
    clazzAgeKVRDD
      .reduceByKey((age1, age2) => {
        // 因为reduceByKey会进行预聚合 所以不适用于求平均值
        val avgAge: Double = (age1 + age2) / 2
        avgAge
      })
      .foreach(println)

  }

}
